use std::time::Duration;

use anyhow::anyhow;
use anyhow::Result;
use clap::Args;
use fudata::db;
use fudata::db::FUDB;
use fudata::model::kline::Kline;
use fudata::model::security::Security;
use futures::StreamExt;
use log::debug;
use log::error;
use log::info;
use surrealdb::engine::remote::ws;
use surrealdb::method::Stream;
use tokio::time::interval;
use tokio::time::Interval;
use zenoh::key_expr::KeyExpr;
use zenoh::prelude::r#async::*;
use zenoh::queryable::Queryable;

use crate::cli::cnf::zenohconf::CommonArgs;
use crate::cli::cnf::LOGO;
use crate::cli::commands::register_ctrl_c;
use crate::cli::commands::wait_until_next_scheduled_time;
use crate::todo::worker_query;
use crate::todo::MAX_WORKERS;
use tokio::select;

#[derive(Args, Debug)]
pub struct SynsCommandArguments {
    #[arg(
        help = "Database path used for storing data",
        short = 'p',
        long = "path",
        requires = "path",
        default_value = "memory"
    )]
    pub path: String,

    #[arg(
        help = "settings to craw securities",
        short = 's',
        long = "settings",
        default_value = "./config.toml"
    )]
    /// settings in toml format, e.g. config.toml
    pub settings: String,
    #[arg(short, long, default_value = "demo/example/**")]
    /// The selection of resources to store.
    key: KeyExpr<'static>,
    #[arg(long)]
    /// Declare the storage as complete w.r.t. the key expression.
    complete: bool,
    #[command(flatten)]
    common: CommonArgs,
}

///
/// https://zenoh.io/docs/overview/zenoh-in-action/
///
/// This animation shows a simple query in action through Zenoh. You can see the presence of storages and queryables.
/// A queryable is any process that can reply to queries. A storage is a combination of a subscriber and a queryable.
///
pub async fn init(SynsCommandArguments {
    path,
    settings,
    key,
    complete,
    common,
}: SynsCommandArguments) -> Result<()> {
    let dbrs = db::init().await;
    match dbrs {
        Ok(_) => {}
        Err(e) => {
            return Err(anyhow!(e));
        }
    }

    info!("Starting fudacraw sync...");

    debug!("(sync::init) arguments: path= {}", path);
    debug!("(sync::init) arguments: settings=  {}", settings);

    info!("Opening session...");

    let _config = zenoh::config::Config::default();
    let config: zenoh::config::Config = common.into();
    let session = zenoh::open(config).res().await.unwrap();

    // Listen for Ctrl-C signal
    let shutdown_rx_arc = register_ctrl_c();
    let query_injector = worker_query::start_worker_pool(MAX_WORKERS );

    let mut main_state = MainState::new(&session).await?;

    loop {
        // Wrap the receiver in a cloneable Arc<Mutex<_>> to allow concurrent access
        let mut shutdown_rx_locked = shutdown_rx_arc.lock().await;

        select!(
            _ = shutdown_rx_locked.recv()  => {
                info!("Shutting down from signal handler");
                break;
            }

            next_time = wait_until_next_scheduled_time("0/5 * * * * *","next_time" )=>{
                info!("\n scheduled time: {:?} ", next_time );
            }

            _logo_time = wait_until_next_scheduled_time("0 0/30 * * * *","_logo_time" )=>{
                println!("{}", LOGO );
            }

            // read local database: provide query security service to remote peers
            queryable_securities = main_state.queryable_securities.recv_async() => {
                if let Ok(query) = queryable_securities {
                    let mut is_other_peer = true; 
                    if let Some(zid) = query.value(){if zid.to_string() == session.zid().to_string(){ is_other_peer=false }};

                    if is_other_peer{
                        // get a query request from remote peer
                        let _=   query_injector.push(query);
                    }
                }
            },

            // read local database: provide query klines service to remote peers
            queryable_klines = main_state.queryable_klines.recv_async() => {
                if let Ok(query) = queryable_klines {
                    let mut is_other_peer = true; 
                    if let Some(zid) = query.value(){if zid.to_string() == session.zid().to_string(){ is_other_peer=false }};

                    if is_other_peer{
                        // get a query request from remote peer
                        let _=   query_injector.push(query);
                    }
                }
            },

            // read local database: publish live security updates to remote peers
            live_update_security = main_state.live_security_stream.next() => {
                if let Some(update) = live_update_security {
                    match update {
                        Ok(notification) => {
                            info!(">> [Live Security Update] New Data: {:?}", notification);

                            // test/test> update security:⟨BJ.OPF.400001⟩ set name='东方龙混合1'

                            // Further processing or publishing of the data
                        },
                        Err(error) => {
                            error!("Error receiving live security update: {:?}", error);
                        },
                    }
                }
            },

            // read local database: publish live kline updates to remote peers
            live_update_kline = main_state.live_kline_stream.next() => {
                if let Some(update) = live_update_kline {
                    match update {
                        Ok(notification) => {
                            info!(">> [Live Kline Update] New Data: {:?}", notification);

                            // test/test> update security:⟨BJ.OPF.400001⟩ set name='东方龙混合1'

                            // Further processing or publishing of the data
                        },
                        Err(error) => {
                            error!("Error receiving live kline update: {:?}", error);
                        },
                    }
                }
            },

            // read local database: publish live kline updates to remote peers
            live_update_tick = main_state.live_tick_stream.next() => {
                if let Some(update) = live_update_tick {
                    match update {
                        Ok(notification) => {
                            info!(">> [Live Tick Update] New Data: {:?}", notification);

                            // test/test> update security:⟨BJ.OPF.400001⟩ set name='东方龙混合1'

                            // Further processing or publishing of the data
                        },
                        Err(error) => {
                            error!("Error receiving live tick update: {:?}", error);
                        },
                    }
                }
            },
        );
    }

    Ok(())
}

pub struct MainState<'a> {
    pub subscriber_demo: zenoh::subscriber::Subscriber<'a, flume::Receiver<Sample>>,
    pub queryable_securities: Queryable<'a, flume::Receiver<zenoh::queryable::Query>>,
    pub queryable_klines: Queryable<'a, flume::Receiver<zenoh::queryable::Query>>,
    pub live_security_stream: Stream<'a, ws::Client, Vec<Security>>,
    pub live_kline_stream: Stream<'a, ws::Client, Vec<Kline>>,
    pub live_tick_stream: Stream<'a, ws::Client, Vec<Kline>>,
    pub timer: Interval,
}

impl<'a> MainState<'a> {
    pub async fn new(session: &'a Session) -> Result<Self> {
        let key = "demo/example/**";
        let complete = false;
        info!("Declaring Subscriber on '{key}'...");
        let subscriber_demmo = session.declare_subscriber(key).res().await.unwrap();

        info!("Declaring Queryable on '{key}'...");

        let queryable_securities = session
            .declare_queryable("securities")
            .complete(complete)
            .res().await
            .unwrap();

        let queryable_klines = session
            .declare_queryable("klines")
            .complete(complete)
            .res().await
            .unwrap();

        let live_security_stream: Stream<ws::Client, Vec<Security>> = FUDB.select(
            "security"
        ).live().await?;
        let live_kline_stream: Stream<ws::Client, Vec<Kline>> = FUDB.select("kline").live().await?;
        let live_tick_stream: Stream<ws::Client, Vec<Kline>> = FUDB.select("tick").live().await?;

        // Create a timer that ticks every x seconds
        let timer: tokio::time::Interval = interval(Duration::from_secs(10));

        Ok(Self {
            subscriber_demo: subscriber_demmo,
            queryable_securities,
            queryable_klines,
            live_security_stream,
            live_kline_stream,
            live_tick_stream,
            timer,
        })
    }
}
